package com.github.binarylei.kafka;

import kafka.producer.KeyedMessage;
import kafka.javaapi.producer.Producer;
import kafka.producer.ProducerConfig;

import java.util.Date;
import java.util.Properties;
import java.util.concurrent.TimeUnit;

/**
 * @author: leigang
 * @version: 2018-04-12
 */
public class KafkaProducer {

    public static void main(String[] args) throws InterruptedException {
        Properties properties = new Properties();

        Properties props = new Properties();
        props.put("zookeeper.connect", "127.0.0.1:2181");
        //配置value的序列化类
        //key的序列化类key.serializer.class可以单独配置，默认使用value的序列化类
        props.put("serializer.class", "kafka.serializer.StringEncoder");
        //配置kafka集群的broker地址，建议配置两个以上，以免其中一个失效，但不需要配全，集群会自动查找leader节点。
        props.put("metadata.broker.list", "127.0.0.1:9092");
        props.put("request.required.acks", "1");

        ProducerConfig config = new ProducerConfig(props);
        Producer<String, String> producer = new Producer<String, String>(config);

        for (int i = 0; i < 10; i++) {
            KeyedMessage<String, String> data = new KeyedMessage<String, String>("test", "test-" + i);
            producer.send(data);
            TimeUnit.SECONDS.sleep(1);
        }
        producer.close();
    }
}
